package com.warren.financial.lease.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.warren.financial.lease.realtime.app.func.AsyncDimFunctionHBase;
import com.warren.financial.lease.realtime.bean.DwsAuditAuditManCancelBean;
import com.warren.financial.lease.realtime.bean.DwsAuditIndLeaseOrgSalesmanApprovalBean;
import com.warren.financial.lease.realtime.util.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @AUTHOR warren
 * @date 2024/1/11
 * @Introduction
 **/
public class DwsAuditAuditManCancelWin {
    public static void main(String[] args) throws Exception {
        // 1. 初始化流处理环境
        String appName = "dws_audit_audit_man_cancel_win";
        StreamExecutionEnvironment env = CreateEnvUtil.getExecutionEnvironment(8082, appName);
        env.setParallelism(1);

        // 2. 从Kafka 目标主题消费数据
        String topic = "financial_dwd_audit_cancel";
        KafkaSource<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(topic, appName, OffsetsInitializer.earliest());
        DataStreamSource<String> kafkaSource = env.fromSource(kafkaConsumer, WatermarkStrategy.noWatermarks(), "kafka_source");

        // 3. 转换数据结构
        // 取消时可能尚未分配信审经办，因此 auditManId 可能为 null，这部分数据是我们不需要的，舍弃
        SingleOutputStreamOperator<DwsAuditAuditManCancelBean> flatMapStream = kafkaSource.flatMap(new FlatMapFunction<String, DwsAuditAuditManCancelBean>() {
            @Override
            public void flatMap(String value, Collector<DwsAuditAuditManCancelBean> out) throws Exception {
                DwsAuditAuditManCancelBean bean = JSON.parseObject(value, DwsAuditAuditManCancelBean.class);
                if (bean.getAuditManId() != null) {
                    bean.setApplyCount(1L);
                    out.collect(bean);
                }
            }
        });

        // 4. 引入水位线
        SingleOutputStreamOperator<DwsAuditAuditManCancelBean> withWaterMark = flatMapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsAuditAuditManCancelBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                .withTimestampAssigner(new SerializableTimestampAssigner<DwsAuditAuditManCancelBean>() {
                    @Override
                    public long extractTimestamp(DwsAuditAuditManCancelBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        //5 按照信审经办ID分组
        KeyedStream<DwsAuditAuditManCancelBean, String> keyByStream = withWaterMark.keyBy(new KeySelector<DwsAuditAuditManCancelBean, String>() {
            @Override
            public String getKey(DwsAuditAuditManCancelBean value) throws Exception {
                return value.getAuditManId();
            }
        });

        //6 开窗
        WindowedStream<DwsAuditAuditManCancelBean, String, TimeWindow> window = keyByStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        // 7 聚合

        SingleOutputStreamOperator<DwsAuditAuditManCancelBean> reduceStream = window.reduce(new ReduceFunction<DwsAuditAuditManCancelBean>() {
            @Override
            public DwsAuditAuditManCancelBean reduce(DwsAuditAuditManCancelBean value1, DwsAuditAuditManCancelBean value2) throws Exception {
                value1.setApplyCount(value1.getApplyCount() + value2.getApplyCount());
                value1.setApplyAmount(value1.getApplyAmount().add(value2.getApplyAmount()));
                return value1;
            }
        }, new ProcessWindowFunction<DwsAuditAuditManCancelBean, DwsAuditAuditManCancelBean, String, TimeWindow>() {
            @Override
            public void process(String s, ProcessWindowFunction<DwsAuditAuditManCancelBean, DwsAuditAuditManCancelBean, String, TimeWindow>.Context context, Iterable<DwsAuditAuditManCancelBean> elements, Collector<DwsAuditAuditManCancelBean> out) throws Exception {
                String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                String curDate = DateFormatUtil.toDate(context.window().getStart());
                for (DwsAuditAuditManCancelBean element : elements) {
                    element.setStt(stt);
                    element.setEdt(edt);
                    element.setCurDate(curDate);
                    out.collect(element);
                }
            }
        });

        reduceStream.print("===============>");

        //8 关联维度信息
        SingleOutputStreamOperator<DwsAuditAuditManCancelBean> auditStream = AsyncDataStream.unorderedWait(reduceStream, new AsyncDimFunctionHBase<DwsAuditAuditManCancelBean>() {
            @Override
            public String getTable() {
                return "dim_employee";
            }

            @Override
            public String getId(DwsAuditAuditManCancelBean bean) {
                return bean.getAuditManId();
            }

            @Override
            public void addDim(DwsAuditAuditManCancelBean bean, JSONObject dim) {
                bean.setAuditManName(dim.getString("name"));
            }
        }, 60, TimeUnit.SECONDS);

        // 9 写入doris
        auditStream.map(new MapFunction<DwsAuditAuditManCancelBean, String>() {
            @Override
            public String map(DwsAuditAuditManCancelBean value) throws Exception {
                return Bean2JSONUtil.bean2JSON(value);
            }
        }).sinkTo(DorisUtil.getDorisSink("financial_lease_realtime.dws_audit_audit_man_cancel_win","dws_audit_audit_man_cancel_win"));

        // 10 执行
        env.execute();
    }
}
